package skewreduce.framework;

import java.util.Timer;
import java.util.TimerTask;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.TaskInputOutputContext;

import skewreduce.lib.KahanSum;


public class ProgressSensor extends TimerTask {
    protected static final Log LOG = LogFactory.getLog("PROGRESS");

    class loopStat {
        long beginTime;
        long endTime;
        long n;
        volatile long i = -1;

        int updates;
        KahanSum average  = new KahanSum();
        KahanSum variance = new KahanSum();
        double avgOld;


        public void begin(long n) {
            this.beginTime = System.currentTimeMillis();
            this.n = n;
            this.i = 0;
        }

        public void end() {
            this.endTime = System.currentTimeMillis();
        }

        public void incr() { ++i; }

        public long get() { return i; }
        public long getLimit() { return n; }

        public boolean isStarted() { return i >= 0; }

        public void update(double v) {
            ++updates;
            average.add(v);
            double avgNew = average.value()/updates;
            variance.add( (v - avgNew) * (v - avgOld) );
            avgOld = avgNew;
        }

        public double getAverage() {
            return average.value() / updates;
        }

        public double getStdev() {
            return Math.sqrt(variance.value()/(updates-1) );
        }
    }

    private final TaskInputOutputContext<?,?,?,?> context;
    private final Timer timer;
    private final int numLoops;
    private final float monitorInterval;

    private loopStat[]   loopInfo;
    private volatile int loopIndex;
    private boolean isGracePeriod;

    private int myLoopIndex;
    private long myI;

    private static final String PROGRESS_INTERVAL_ATTR = "skewreduce.monitoring.job.progress";
    private static final String GRACE_PERIOD_ATTR = "skewreduce.monitoring.job.graceperiod";
    private static final String NUM_OF_LOOPS = "skewreduce.monitoring.num.loops";

    public ProgressSensor(TaskInputOutputContext<?,?,?,?> context) {
        this.context = context;
        this.timer = new Timer(true);
        
        Configuration conf = context.getConfiguration();
        monitorInterval = conf.getFloat(PROGRESS_INTERVAL_ATTR,3.0f);
        long gracePeriod = conf.getLong(GRACE_PERIOD_ATTR,2*60) * 1000;

        numLoops = conf.getInt(NUM_OF_LOOPS,0);
        if ( numLoops == 0 )
            throw new IllegalArgumentException("Can not determine the number of loops");

        this.loopInfo = new loopStat[numLoops];
        for ( int i = 0; i < numLoops; ++i )
            loopInfo[i] = new loopStat();

        timer.scheduleAtFixedRate(this,0,(long)(monitorInterval*1000));
        timer.schedule(new SetGracePeriodOver(), gracePeriod);
        // report every minute
        timer.scheduleAtFixedRate(new ReportProgress(),0,60*1000);
    }

    public void beginLoop(long n) {
        loopInfo[loopIndex].begin(n);
    }

    public void endLoop() {
        loopInfo[loopIndex].end();
        ++loopIndex;
    }

    public void incrLoop() {
        loopInfo[loopIndex].incr();
    }

    public void cleanup() {
        // dump all timing information
        int i = 0;
        for ( loopStat ls : loopInfo ) {
            // idx, begin, end, duration, i, n, avg, stdev
            //
            LOG.info(String.format("T,%d,%d,%d,%d,%d,%d,%f,%f",i++,ls.beginTime,
                        ls.endTime,ls.endTime - ls.beginTime,
                        ls.i,
                        ls.n,
                        ls.getAverage(),
                        ls.getStdev()
            ));
        }
    }

    public void run() {
        // make a copy of volatile variable
        int copyLoopIndex = loopIndex;
        long I = loopInfo[myLoopIndex].get();

        if ( copyLoopIndex == myLoopIndex ) {
            // we are seeing the same loop from the previous time
            if ( loopInfo[myLoopIndex].isStarted() ) {

                double ips = (I - myI) / monitorInterval;
                loopInfo[myLoopIndex].update(ips);

                LOG.info(String.format("R,%d,%d,%d,%f",myLoopIndex,I,loopInfo[myLoopIndex].n - I,ips));
                //,loopInfo[myLoopIndex].getAverage(),loopInfo[myLoopIndex].getStdev()));

                if ( ! isGracePeriod ) {
                    // FIXME: do some action?
                }
            } else {
                // not yet started. do nothing
            }
        } else {
            // just completed a loop.
            // the last one is usually garbage so discard.
            myLoopIndex = copyLoopIndex;    // we are seeing a new counter
            I = ( myLoopIndex < numLoops ) ? loopInfo[myLoopIndex].get() : 0;
            // FIXME: array index out of bounds exception
        }

        myI = I;
       
        if ( myLoopIndex >= numLoops ) {
            cancel();   // cancel schedule
        }
    }

    class SetGracePeriodOver extends TimerTask {
        public void run() {
            LOG.info("grace period is over");
            isGracePeriod = false;
        }
    }

    class ReportProgress extends TimerTask {
        public void run() {
            context.progress();
        }
    }
}
